package net.bwie.douyin.dwd.log1.Function;

import net.bwie.douyin.dwd.log1.bean.RoomInteraction;
import net.bwie.douyin.dwd.log1.bean.TikTokBean;
import org.apache.flink.api.common.functions.AggregateFunction;

/**
 * @ClassName：InteractionAggregator
 * @Author: ZddddQ
 * @Date: 2025/5/28 19:24
 * @Description: 必须描述类做什么事情, 实现什么功能
 */

import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 优化版实时互动统计功能
 * 使用增量聚合+窗口元数据组合
 */
public class RealTimeInteractionOptimized {

    // 增量聚合函数 - 实时统计各类互动事件
    public static class InteractionAggregator
            implements AggregateFunction<TikTokBean, InteractionAccumulator, InteractionAccumulator> {

        @Override
        public InteractionAccumulator createAccumulator() {
            return new InteractionAccumulator();
        }

        @Override
        public InteractionAccumulator add(TikTokBean bean, InteractionAccumulator accumulator) {
            // 根据事件类型更新对应的计数器
            switch (bean.getEventType()) {
                case "like":
                    accumulator.likeCount++;
                    break;
                case "share":
                    accumulator.shareCount++;
                    break;
                case "message":
                    accumulator.messageCount++;
                    break;
                // 其他事件类型可以忽略或处理
            }
            // 更新总互动数（点赞+分享+消息）
            accumulator.interactionCount =
                    accumulator.likeCount + accumulator.shareCount + accumulator.messageCount;

            return accumulator;
        }

        @Override
        public InteractionAccumulator getResult(InteractionAccumulator accumulator) {
            return accumulator;
        }

        @Override
        public InteractionAccumulator merge(InteractionAccumulator a, InteractionAccumulator b) {
            // 合并两个累加器
            a.likeCount += b.likeCount;
            a.shareCount += b.shareCount;
            a.messageCount += b.messageCount;
            a.interactionCount = a.likeCount + a.shareCount + a.messageCount;
            return a;
        }
    }

    // 窗口处理函数 - 添加窗口信息
    public static class InteractionWindowProcessor
            extends ProcessWindowFunction<InteractionAccumulator, RoomInteraction, String, TimeWindow> {

        private final FastDateFormat fastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

        @Override
        public void process(
                String roomId,
                Context context,
                Iterable<InteractionAccumulator> aggregates,
                Collector<RoomInteraction> out) {

            // 获取聚合结果
            InteractionAccumulator accumulator = aggregates.iterator().next();

            // 获取窗口元数据
            TimeWindow window = context.window();
            String windowStart = fastDateFormat.format(window.getStart());
            String windowEnd = fastDateFormat.format(window.getEnd());

            // 输出带窗口信息的最终结果
            out.collect(new RoomInteraction(
                    windowStart,
                    windowEnd,
                    roomId,
                    accumulator.interactionCount,
                    accumulator.shareCount,
                    accumulator.likeCount,
                    accumulator.messageCount,
                    System.currentTimeMillis()
            ));
        }
    }

    // 自定义累加器类，用于存储中间状态
    public static class InteractionAccumulator {
        public int interactionCount = 0;  // 总互动数
        public int shareCount = 0;        // 分享数
        public int likeCount = 0;         // 点赞数
        public int messageCount = 0;      // 消息数
    }
}
